package com.chukun.flink.state;

import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.queryablestate.client.QueryableStateClient;

import java.net.UnknownHostException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @author chukun
 * @version 1.0.0
 * @description flink查询作业运行过程中一些状态数据
 * @createTime 2022年06月04日 12:14:00
 */
public class QueryStateClient {

    /**
     * 更多 queryable 操作详见: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/fault-tolerance/queryable_state/
     * @param args
     * @throws UnknownHostException
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void main(String[] args) throws UnknownHostException, ExecutionException, InterruptedException {

        String jobId = "488266c6a81ca05fe7557ad7865aa789";

        // 创建查询客户端
        QueryableStateClient queryableStateClient = new QueryableStateClient("linux01",9069);

        // 创建状态数据
        ValueStateDescriptor<Tuple2<Integer, Integer>> stateDescriptor =
                new ValueStateDescriptor<>(
                        "ValueState",
                        TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {
                        }));

        // 获取状态数据
        ValueState<Tuple2<Integer, Integer>> state = getState(jobId, queryableStateClient, stateDescriptor);

        System.out.println("查询valuestate : " + state);


    }

    /**
     * 获取状态信息
     * @param jobId
     * @param client
     * @param stateDescriptor
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    private static ValueState<Tuple2<Integer, Integer>> getState(
            String jobId,
            QueryableStateClient client,
            ValueStateDescriptor<Tuple2<Integer, Integer>> stateDescriptor) throws ExecutionException, InterruptedException {
        // 从 kvstate中获取状态数据
        CompletableFuture<ValueState<Tuple2<Integer, Integer>>> stateFuture = client.getKvState(
                JobID.fromHexString(jobId),
                "local-client-queryable",
                null,
                new TypeHint<Tuple2<Integer, String>>() {
                },
                stateDescriptor);
        return stateFuture.get();
    }
}
